package com.fivefu.core.skyeye.event.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fivefu.core.skyeye.event.entity.DbAiRedisStream;
import com.fivefu.core.skyeye.event.entity.DbAiScene;
import com.fivefu.core.skyeye.event.properties.RedisStringStreamBean;
import com.fivefu.core.skyeye.event.properties.StreamBean;
import com.fivefu.core.skyeye.event.services.DbAiRedisStreamService;
import com.fivefu.core.skyeye.event.services.DbAiSceneService;
import com.fivefu.core.skyeye.event.services.RedisStreamService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class MapRedisStreamListener implements StreamListener<String, MapRecord<String, String, String>> {

    private Logger logger = LoggerFactory.getLogger(MapRedisStreamListener.class);

    @Autowired
    private RedisStringStreamBean stringStreamBean;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private DbAiRedisStreamService dbAiRedisStreamService;
    @Autowired
    private RedisStreamService redisStreamService;
    @Autowired
    private DbAiSceneService dbAiSceneService;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        logger.info("监听到 redis 信息");

        // 消息ID
        RecordId messageId = message.getId();//RecordID
        String streamName = message.getStream();//streamName
        // 消息的key和value
        String streamValue =  JSON.toJSONString(message.getValue());//streamValue

        // 通过RedisTemplate手动确认消息
        List<StreamBean> streamList = stringStreamBean.getMapRedis();
        boolean streamFlag = false;
        for (StreamBean redisStream : streamList) {
            if (redisStream.getStreamName().equals(streamName)){
                this.stringRedisTemplate.opsForStream().acknowledge(redisStream.getStreamGroup(), message);
                streamFlag = true;
            }
        }
        if (!streamFlag){
            return;
        }

        //redis信息
        DbAiRedisStream redisStream = dbAiRedisStreamService.queryRedisStreamByRecordID(messageId.toString());
        Long id = null;
        boolean synFlag = false;
        if (redisStream == null){
            id = dbAiRedisStreamService.insertRedisStream(messageId.toString(), streamName, streamValue);
        }else {
            id = redisStream.getId();
            if (redisStream.getStatus() != null && redisStream.getStatus().equals(1)){//同步成功
                synFlag = true;
            }
        }
        if (!synFlag){
            if ("mimir_event".equals(streamName) || "mimir".equals(streamName)){//事件
                JSONObject messageObj = JSON.parseObject(streamValue);

                String type = messageObj.getString("mq_type");
                Integer cid = messageObj.getInteger("cid");
                Long eventid = messageObj.getLong("event_id");

                //补充redis消息类型、违停类型及案卷编号
                dbAiRedisStreamService.updateRedisStream(id, type, cid, eventid);

                DbAiScene scene = dbAiSceneService.querySceneByCvId(cid);
                if (scene != null){
                    //处理案卷信息
                    boolean flag = redisStreamService.handleMimirEventInfo(scene.getCode(),messageObj);
                    //修改redis数据状态
                    Integer status = 0;
                    if (flag){
                        status = 1;
                    }
                    dbAiRedisStreamService.updateRedisStreamStatus(id, status);
                }
            }
        }







        /*DbAiRedisStream redisStream = dbAiRedisStreamService.queryRedisStreamByRecordID(messageId.toString());
        Long id = null;
        if (redisStream == null){
            id = dbAiRedisStreamService.insertRedisStream(messageId.toString(), stream, string);
        }else{
            id = redisStream.getId();
        }
        boolean flag = stringRedisStreamService.handleMapRedisStreamMessage(messageId, stream, string);

        Integer status = 0;
        if (flag){
            status = 1;
        }
        dbAiRedisStreamService.updateRedisStreamStatus(id, status);*/
    }

}
